package com.demo.rocketmq.consumer;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.demo.rocketmq.annotation.MqConsumer;
import com.demo.rocketmq.constants.TopicConstants;
import com.demo.rocketmq.entity.RetrySendMsg;
import com.demo.rocketmq.mapper.RetrySendMsgMapper;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

/**
 * 消费者消费消息后自动发送回调消息给生产者，生产者收到回调消息后确认消息已被消费者消息
 */
@Slf4j
@Service
public class CallbackRocketMqConsumer {

    @Resource
    private RetrySendMsgMapper retrySendMsgMapper;

    /**
     * 消费消息，并向主题：test-topic-1-callback 自动发送回调消息
     * @param msg
     */
    @MqConsumer(topic = TopicConstants.TEST_TOPIC_1, consumerGroup = "mq_consumer_callback_group",
            callbackTopic = "test-topic-1-callback")
    public void callback(String msg) {
        log.info("测试回调消息：" + msg);
    }

    /**
     * 接收主题：test-topic-1-callback 消息，更新消息表数据
     * @param callbackKey
     */
    @MqConsumer(topic = TopicConstants.TEST_TOPIC_1_CALLBACK, consumerGroup = "mq_consumer_update_callback_group")
    public void consumeCallback(String callbackKey) {
        log.info("接收回调callbakkey：{}", callbackKey);
        LambdaQueryWrapper<RetrySendMsg> wrapper = new LambdaQueryWrapper();
        wrapper.eq(RetrySendMsg::getCallbackKey, callbackKey);

        RetrySendMsg retrySendMsg = new RetrySendMsg();
        retrySendMsg.setFlag(1);

        retrySendMsgMapper.update(retrySendMsg, wrapper);
    }
}
